package com.example.user.rocketmq.test;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * @Author 刘武贵
 * @Date 2024-03-10 17:30
 * @Description 功能描述：RocketMQ消费者
 */
public class RocketMQConsumer {

    /**
     * 主函数：创建并启动一个RocketMQ消息消费者，用于订阅并消费消息。
     * @param args 命令行参数
     * @throws Exception 抛出异常的条件不明确，因为捕获了所有异常
     */
    public static void main(String[] args) throws Exception {
        try{
            // 初始化消息消费者
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("bingheConsumerGroup");
            // 设置NameServer地址，用于连接RocketMQ服务
            consumer.setNamesrvAddr("127.0.0.1:9876");
            // 订阅指定的主题，*表示接受该主题下所有标签的消息
            consumer.subscribe("bingheTopic", "*");
            // 注册消息监听器，当有消息到达时，执行该监听器的回调方法
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list,
                                                                ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                    // 处理接收到的消息，并返回消费状态
                    // 打印消息内容
                    System.out.println("消费者收到的消息为：" + list);
                    // 返回消费成功状态
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            // 启动消费者实例
            consumer.start();
            // 输出启动成功的提示信息
            System.out.println("消费者启动成功");
        }catch (Exception e){
            // 捕获并打印所有异常
            e.printStackTrace();
        }
    }
}
